feat: WebSocket transport — full-duplex resumable streams#969
Open
AlemTuzlak wants to merge 24 commits into
Open
feat: WebSocket transport — full-duplex resumable streams#969AlemTuzlak wants to merge 24 commits into
AlemTuzlak wants to merge 24 commits into
Conversation
Add a full-duplex WebSocket chat demo to ts-react-chat: a Vite dev-server
plugin (websocket-chat-plugin.ts) hooks the Node http server's `upgrade`
event and wires toWebSocketStream/resumeWebSocketStream around chat() with
gpt-5.5, mirroring the pattern in testing/e2e's durable-delivery-ws-plugin
(no WebSocketPair on Node/Nitro). The /websocket-chat route uses useChat with
the webSocket() connection adapter, linked from the nav.
@tanstack/ai-react didn't re-export webSocket/WebSocketConnectionOptions from
@tanstack/ai-client yet (only the other connection adapters were), so this
adds that re-export as it's required for the example's `import { useChat,
webSocket } from '@tanstack/ai-react'`.
Document the new WebSocket transport (toWebSocketStream/toWebSocketResponse, resumeWebSocketStream/resumeWebSocketResponse, client webSocket()): a new standalone WebSockets page, a short intro + snippet in the Overview, a reconnect/lifecycle summary in Advanced, and a pointer from Connection Adapters to the built-in adapter instead of the old hand-rolled example.
…/heartbeat; remove abort listener leak
Contributor
🚀 Changeset Version Preview8 package(s) bumped directly, 39 bumped as dependents. 🟥 Major bumps
🟨 Minor bumps
🟩 Patch bumps
|
|
View your CI Pipeline Execution ↗ for commit 2225a78
☁️ Nx Cloud last updated this comment at |
Contributor
📝 WalkthroughWalkthroughA new resumable WebSocket transport adds server streaming and replay helpers, a client adapter with reconnect tracking, framework re-exports, React and end-to-end examples, comprehensive tests, and documentation covering protocol, lifecycle, hosting, and durability. ChangesWebSocket transport
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant useChat
participant webSocket
participant toWebSocketStream
participant chat
participant memoryStream
useChat->>webSocket: send RunAgentInput
webSocket->>toWebSocketStream: send run frame
toWebSocketStream->>chat: invoke onRun with turn context
chat->>memoryStream: persist durable chunks
toWebSocketStream-->>webSocket: send chunk envelopes
webSocket-->>useChat: yield streamed chunks
webSocket->>toWebSocketStream: reconnect with runId and offset
toWebSocketStream->>memoryStream: replay missing chunks
memoryStream-->>useChat: deliver resumed chunks
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 ESLint
ESLint install timed out. The project may have too many dependencies for the sandbox. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
@tanstack/ai
@tanstack/ai-acp
@tanstack/ai-angular
@tanstack/ai-anthropic
@tanstack/ai-bedrock
@tanstack/ai-claude-code
@tanstack/ai-client
@tanstack/ai-code-mode
@tanstack/ai-code-mode-skills
@tanstack/ai-codex
@tanstack/ai-devtools-core
@tanstack/ai-durable-stream
@tanstack/ai-elevenlabs
@tanstack/ai-event-client
@tanstack/ai-fal
@tanstack/ai-gemini
@tanstack/ai-grok
@tanstack/ai-grok-build
@tanstack/ai-groq
@tanstack/ai-isolate-cloudflare
@tanstack/ai-isolate-node
@tanstack/ai-isolate-quickjs
@tanstack/ai-mcp
@tanstack/ai-mistral
@tanstack/ai-ollama
@tanstack/ai-openai
@tanstack/ai-opencode
@tanstack/ai-openrouter
@tanstack/ai-preact
@tanstack/ai-react
@tanstack/ai-react-ui
@tanstack/ai-sandbox
@tanstack/ai-sandbox-cloudflare
@tanstack/ai-sandbox-daytona
@tanstack/ai-sandbox-docker
@tanstack/ai-sandbox-local-process
@tanstack/ai-sandbox-sprites
@tanstack/ai-sandbox-vercel
@tanstack/ai-solid
@tanstack/ai-solid-ui
@tanstack/ai-svelte
@tanstack/ai-utils
@tanstack/ai-vue
@tanstack/ai-vue-ui
@tanstack/openai-base
@tanstack/preact-ai-devtools
@tanstack/react-ai-devtools
@tanstack/solid-ai-devtools
commit: |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (2)
examples/ts-react-chat/src/lib/websocket-chat-plugin.ts (1)
40-46: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePropagate the abort reason to the new controller.
When bridging an
AbortSignalto a newAbortController, passingsignal.reasonensures that any downstream cancellation logs or error handlers receive the correct context rather than a generic abort error.
examples/ts-react-chat/src/lib/websocket-chat-plugin.ts#L40-L46: passsignal.reasontocontroller.abort()in the plugin.docs/resumable-streams/websockets.md#L56-L61: passsignal.reasontocontroller.abort()in the documentation sample.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/ts-react-chat/src/lib/websocket-chat-plugin.ts` around lines 40 - 46, Propagate the original abort reason when bridging signals: update abortControllerFromSignal in examples/ts-react-chat/src/lib/websocket-chat-plugin.ts lines 40-46 and the corresponding abort listener in docs/resumable-streams/websockets.md lines 56-61 to pass signal.reason to controller.abort(), preserving the existing immediate and event-driven cancellation behavior.packages/ai/src/stream-to-websocket.ts (1)
204-211: 🚀 Performance & Scalability | 🔵 Trivial | ⚖️ Poor tradeoffNo send backpressure —
bufferedAmountis declared but never consulted.Both pump loops call
socket.send(...)on every chunk without awaiting drain. A fastonRun/durable replay against a slow client will queue frames unbounded in the socket's send buffer, growing memory. TheWebSocketLike.bufferedAmountfield (Line 20) appears intended for this but is unused; the resume pump at Line 265-266 has the same gap. Consider pausing iteration whilebufferedAmountexceeds a high-water mark.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/ai/src/stream-to-websocket.ts` around lines 204 - 211, Update both stream pump loops around source and init.onRun, plus the resume pump, to apply send backpressure before calling socket.send. Pause iteration while WebSocketLike.bufferedAmount exceeds a defined high-water threshold, then resume once the buffer drains, preserving chunk order and existing frame encoding.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/chat/connection-adapters.md`:
- Around line 299-312: Add a concise server endpoint example to the WebSockets
section alongside the existing client webSocket usage, demonstrating the paired
toWebSocketStream or toWebSocketResponse integration. Keep the example
consistent with the documented server-side WebSocket APIs and retain the
existing link for protocol and hosting details.
In `@docs/resumable-streams/websockets.md`:
- Line 243: Remove the `as unknown as WebSocketLike` assertion from the
`socketLike` assignment in the WebSocket example; pass `ws` directly if it
satisfies the interface, otherwise use a type guard or property validation so
the sample type-checks without any `as` cast.
In `@packages/ai-client/src/connection-adapters.ts`:
- Around line 1690-1691: Update the ws.onmessage handler to guard
JSON.parse(String(event.data)) with try/catch. On malformed frames, call failAll
with a StreamReadError containing the caught error so stream consumers receive a
deterministic failure instead of hanging; preserve the existing handling for
successfully parsed messages.
- Around line 1721-1734: Update the ws.onclose handler in the connection adapter
to notify joinRun listeners when the socket closes even if currentSession is
undefined. Preserve the existing session reconnect and non-durable failure
behavior, while ensuring pending joinRun iterators are rejected or otherwise
terminated instead of remaining unresolved.
In `@packages/ai/src/stream-to-websocket.ts`:
- Around line 179-221: The handleInbound function must not overwrite an active
controller for the same runId. Before activeTurns.set(params.runId, turnAbort),
retrieve any existing controller, abort it or reject the duplicate run, then
store the new controller only when appropriate; preserve the existing ownership
check in the finally cleanup.
---
Nitpick comments:
In `@examples/ts-react-chat/src/lib/websocket-chat-plugin.ts`:
- Around line 40-46: Propagate the original abort reason when bridging signals:
update abortControllerFromSignal in
examples/ts-react-chat/src/lib/websocket-chat-plugin.ts lines 40-46 and the
corresponding abort listener in docs/resumable-streams/websockets.md lines 56-61
to pass signal.reason to controller.abort(), preserving the existing immediate
and event-driven cancellation behavior.
In `@packages/ai/src/stream-to-websocket.ts`:
- Around line 204-211: Update both stream pump loops around source and
init.onRun, plus the resume pump, to apply send backpressure before calling
socket.send. Pause iteration while WebSocketLike.bufferedAmount exceeds a
defined high-water threshold, then resume once the buffer drains, preserving
chunk order and existing frame encoding.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 9e15cc23-dd08-43f6-a7e2-eebf70c11658
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (30)
.changeset/websocket-transport.mddocs/chat/connection-adapters.mddocs/config.jsondocs/resumable-streams/advanced.mddocs/resumable-streams/overview.mddocs/resumable-streams/websockets.mdexamples/ts-react-chat/package.jsonexamples/ts-react-chat/src/components/Header.tsxexamples/ts-react-chat/src/lib/websocket-chat-plugin.tsexamples/ts-react-chat/src/routeTree.gen.tsexamples/ts-react-chat/src/routes/websocket-chat.tsxexamples/ts-react-chat/vite.config.tspackages/ai-angular/src/index.tspackages/ai-client/src/connection-adapters.tspackages/ai-client/src/index.tspackages/ai-client/tests/connection-adapters-resumable.test.tspackages/ai-client/tests/connection-adapters-websocket.test.tspackages/ai-react/src/index.tspackages/ai-solid/src/index.tspackages/ai-svelte/src/index.tspackages/ai-vue/src/index.tspackages/ai/src/index.tspackages/ai/src/stream-to-response.tspackages/ai/src/stream-to-websocket.tspackages/ai/tests/stream-to-websocket.test.tstesting/e2e/package.jsontesting/e2e/src/lib/durable-delivery-ws-plugin.tstesting/e2e/src/routes/api.durable-delivery.tstesting/e2e/tests/websocket.spec.tstesting/e2e/vite.config.ts
Comment on lines
+299
to
+312
| ## WebSockets | ||
|
|
||
| For a persistent, resumable WebSocket, use the built-in `webSocket()` adapter instead of hand-rolling a `SubscribeConnectionAdapter`. It opens one socket for the whole conversation, reconnects a dropped durable run automatically, and pairs with the server's `toWebSocketStream` / `toWebSocketResponse`: | ||
|
|
||
| ```typescript | ||
| import { useChat, webSocket } from "@tanstack/ai-react"; | ||
|
|
||
| const connection = webSocket("/api/chat-ws"); | ||
|
|
||
| const { messages, sendMessage } = useChat({ connection }); | ||
| ``` | ||
|
|
||
| See [WebSockets](../resumable-streams/websockets) for the server side, the wire protocol, reconnect details, and hosting on Node vs Cloudflare. | ||
|
|
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add a server endpoint snippet for WebSockets.
As per coding guidelines, when a documentation page covers both server and client behavior, snippets for both halves (the server endpoint and the client consumption) must be included. The SSE section on this page includes a server snippet, but the new WebSockets section only provides the client webSocket snippet. Please add a brief example of the paired server endpoint (e.g., using toWebSocketStream or toWebSocketResponse).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/chat/connection-adapters.md` around lines 299 - 312, Add a concise
server endpoint example to the WebSockets section alongside the existing client
webSocket usage, demonstrating the paired toWebSocketStream or
toWebSocketResponse integration. Keep the example consistent with the documented
server-side WebSocket APIs and retain the existing link for protocol and hosting
details.
Source: Coding guidelines
|
|
||
| wss.handleUpgrade(req, socket, head, (ws) => { | ||
| const request = new Request(url) | ||
| const socketLike = ws as unknown as WebSocketLike |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Remove the as type-assertion cast.
As per coding guidelines, do not use as type-assertion casts in documentation code samples. Examples must type-check without as SomeType; consider using a type guard or validating properties if the types do not natively align, or simply passing ws directly if it structurally satisfies the interface.
🛠 Proposed fix
- const socketLike = ws as unknown as WebSocketLike
-
if (url.searchParams.get('offset') !== null) {
- resumeWebSocketStream(socketLike, { adapter: memoryStream(request) })
+ resumeWebSocketStream(ws, { adapter: memoryStream(request) })
} else {
- handleChatSocket(socketLike, request)
+ handleChatSocket(ws, request)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| const socketLike = ws as unknown as WebSocketLike | |
| if (url.searchParams.get('offset') !== null) { | |
| resumeWebSocketStream(ws, { adapter: memoryStream(request) }) | |
| } else { | |
| handleChatSocket(ws, request) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/resumable-streams/websockets.md` at line 243, Remove the `as unknown as
WebSocketLike` assertion from the `socketLike` assignment in the WebSocket
example; pass `ws` directly if it satisfies the interface, otherwise use a type
guard or property validation so the sample type-checks without any `as` cast.
Source: Coding guidelines
Comment on lines
+1690
to
+1691
| ws.onmessage = (event: MessageEvent) => { | ||
| const parsed: unknown = JSON.parse(String(event.data)) |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="packages/ai-client/src/connection-adapters.ts"
echo "== outline =="
ast-grep outline "$file" --view expanded || true
echo
echo "== relevant slice around lines 1660-1715 =="
sed -n '1660,1715p' "$file" | cat -n
echo
echo "== search for failAll and StreamReadError in file =="
rg -n "failAll|StreamReadError|onmessage|JSON\.parse" "$file"Repository: TanStack/ai
Length of output: 8433
Guard malformed WebSocket frames before parsing
ws.onmessage parses event.data unguarded, so a bad frame throws out of the handler and leaves the stream without a deterministic failure path. Wrap this in try/catch and call failAll(new StreamReadError(error)) (or drop the frame explicitly) so consumers don’t hang on a corrupted message.
Suggested guard
ws.onmessage = (event: MessageEvent) => {
- const parsed: unknown = JSON.parse(String(event.data))
+ let parsed: unknown
+ try {
+ parsed = JSON.parse(String(event.data))
+ } catch (error) {
+ failAll(new StreamReadError(error))
+ return
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| ws.onmessage = (event: MessageEvent) => { | |
| const parsed: unknown = JSON.parse(String(event.data)) | |
| ws.onmessage = (event: MessageEvent) => { | |
| let parsed: unknown | |
| try { | |
| parsed = JSON.parse(String(event.data)) | |
| } catch (error) { | |
| failAll(new StreamReadError(error)) | |
| return | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/ai-client/src/connection-adapters.ts` around lines 1690 - 1691,
Update the ws.onmessage handler to guard JSON.parse(String(event.data)) with
try/catch. On malformed frames, call failAll with a StreamReadError containing
the caught error so stream consumers receive a deterministic failure instead of
hanging; preserve the existing handling for successfully parsed messages.
Comment on lines
+1721
to
+1734
| ws.onclose = () => { | ||
| const session = currentSession | ||
| if (!session || session.signal?.aborted || session.sawTerminal) return | ||
| const lastEventId = session.tracker.lastEventId | ||
| if (lastEventId === undefined) { | ||
| // Non-durable run (no offset ever observed) — nothing to resume | ||
| // from. Surface a hard failure rather than silently reconnecting | ||
| // forever against a server that never tags its events. | ||
| currentSession = undefined | ||
| failAll(new StreamReadError(new Error('WebSocket connection closed'))) | ||
| return | ||
| } | ||
| void reconnect(session, lastEventId) | ||
| } |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the file structure first
ast-grep outline packages/ai-client/src/connection-adapters.ts --view expanded || true
# Inspect the relevant region around the cited lines
sed -n '1600,1785p' packages/ai-client/src/connection-adapters.ts
# Find joinRun / send / subscribe / onclose usage
rg -n "joinRun|subscribe|send\(|onclose|currentSession|sawTerminal|lastEventId|reconnect" packages/ai-client/src/connection-adapters.tsRepository: TanStack/ai
Length of output: 19758
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the runtime contract around the joinRun API and session lifecycle
sed -n '1,260p' packages/ai-client/src/connection-adapters.ts
# Look for tests or callers that clarify intended close behavior
rg -n "joinRun\\(|WebSocket connection closed|failAll\\(|abortSignal|StreamReadError|currentSession = undefined" packages -g '!**/dist/**'Repository: TanStack/ai
Length of output: 34747
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Read the specific implementation in smaller chunks
cat -n packages/ai-client/src/connection-adapters.ts | sed -n '1,260p'Repository: TanStack/ai
Length of output: 11228
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the subscribe() iterator and joinRun() implementation in the WebSocket adapter
sed -n '1770,1915p' packages/ai-client/src/connection-adapters.ts
# Inspect websocket tests around joinRun and close behavior
sed -n '1,260p' packages/ai-client/tests/connection-adapters-websocket.test.ts
sed -n '260,420p' packages/ai-client/tests/connection-adapters-websocket.test.tsRepository: TanStack/ai
Length of output: 20365
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
text = Path("packages/ai-client/src/connection-adapters.ts").read_text()
start = text.index(" subscribe(abortSignal?: AbortSignal): AsyncIterable<StreamChunk> {")
end = text.index("export function stream(", start)
print(text[start:end])
PYRepository: TanStack/ai
Length of output: 39994
Fail joinRun() listeners on socket close
ws.onclose returns early when currentSession is undefined, so a joinRun()-only socket drop can leave the iterator parked on its pending promise forever. Surface the close to join-run listeners too, even without an active send/session.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/ai-client/src/connection-adapters.ts` around lines 1721 - 1734,
Update the ws.onclose handler in the connection adapter to notify joinRun
listeners when the socket closes even if currentSession is undefined. Preserve
the existing session reconnect and non-durable failure behavior, while ensuring
pending joinRun iterators are rejected or otherwise terminated instead of
remaining unresolved.
Comment on lines
+179
to
+221
| async function handleInbound(input: unknown): Promise<void> { | ||
| const params = await chatParamsFromRequestBody(input) | ||
| const turnAbort = new AbortController() | ||
| activeTurns.set(params.runId, turnAbort) | ||
| const ctx: WsRunContext = { | ||
| messages: params.messages, | ||
| threadId: params.threadId, | ||
| runId: params.runId, | ||
| forwardedProps: params.forwardedProps, | ||
| resumeOffset: null, | ||
| request: buildTurnRequest(request, params.runId, null), | ||
| signal: turnAbort.signal, | ||
| } | ||
| try { | ||
| if (init.durability) { | ||
| const adapter = init.durability(ctx) | ||
| const { source, getId } = durableStreamSource( | ||
| init.onRun(ctx), | ||
| adapter, | ||
| { | ||
| abortController: turnAbort, | ||
| ...(init.batch === undefined ? {} : { batch: init.batch }), | ||
| logger, | ||
| }, | ||
| ) | ||
| for await (const chunk of source) { | ||
| socket.send(encodeWsFrame(chunk, getId(chunk))) | ||
| } | ||
| } else { | ||
| for await (const chunk of init.onRun(ctx)) { | ||
| socket.send(encodeWsFrame(chunk, undefined)) | ||
| } | ||
| } | ||
| } finally { | ||
| // Only delete if this turn still owns the entry: a duplicate in-flight | ||
| // runId (e.g. a client resubmitting before the first turn finished) | ||
| // would otherwise let the OLDER turn's cleanup delete the NEWER turn's | ||
| // still-active controller (TOCTOU). | ||
| if (activeTurns.get(params.runId) === turnAbort) { | ||
| activeTurns.delete(params.runId) | ||
| } | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
set -euo pipefail
# Map the relevant file structure and locate the symbols.
git ls-files packages/ai/src | sed -n '1,200p'
printf '\n--- outline ---\n'
ast-grep outline packages/ai/src/stream-to-websocket.ts --view expanded || true
printf '\n--- matches ---\n'
rg -n "activeTurns|runId|close handler|AbortController|handleInbound|onRun|durability" packages/ai/src/stream-to-websocket.ts packages/ai/src -SRepository: TanStack/ai
Length of output: 28059
🏁 Script executed:
set -euo pipefail
# Read the relevant sections with line numbers.
sed -n '1,320p' packages/ai/src/stream-to-websocket.ts | cat -nRepository: TanStack/ai
Length of output: 13582
🏁 Script executed:
set -euo pipefail
# Focused read of the file sections likely relevant to the duplicate-runId concern.
sed -n '120,260p' packages/ai/src/stream-to-websocket.ts | cat -nRepository: TanStack/ai
Length of output: 5819
Do not overwrite an in-flight runId. A second run frame with the same runId replaces the existing AbortController in activeTurns, so the earlier turn becomes unreachable. Both the close handler and the abort control frame only act on the map entry, which means the orphaned turn can keep running until it finishes on its own. Abort or reject any existing controller before storing the new one.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/ai/src/stream-to-websocket.ts` around lines 179 - 221, The
handleInbound function must not overwrite an active controller for the same
runId. Before activeTurns.set(params.runId, turnAbort), retrieve any existing
controller, abort it or reject the duplicate run, then store the new controller
only when appropriate; preserve the existing ownership check in the finally
cleanup.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
WebSocket transport
Adds WebSockets as a third streaming transport alongside SSE and NDJSON, reusing the delivery-durability seam that landed in #955. Full-duplex, conversation-scoped, resumable.
Server (
@tanstack/ai)toWebSocketStream(socket, request, { onRun, durability?, batch?, heartbeatMs?, idleTimeoutMs?, debug? })— portable core that pumps a conversation over an already-accepted WHATWGWebSocketLikeserver socket (Node viaws, Bun, etc.).toWebSocketResponse(request, init)— thin wrapper that upgrades viaWebSocketPairand returns a 101Responseon Cloudflare Workers/Durable Objects; throws elsewhere, pointing you totoWebSocketStream.resumeWebSocketStream(socket, { adapter })/resumeWebSocketResponse({ adapter })— read-only replay of a run from the durability log (no model call).chat()turns (client-tool resubmits, follow-up messages), you pass anonRun(ctx) => AsyncIterable<StreamChunk>factory instead of a prebuilt stream — it's called per inboundRunAgentInputframe. Durability is keyed per turn and reuses the existingdurableStreamSource(now exported), so server→client frames carry the same{ id, chunk }envelope as NDJSON.{ type: 'abort', runId }frame (aborts only that turn), or idle timeout; periodic{ type: 'ping' }heartbeat.Client (
@tanstack/ai-client, re-exported fromai-react/-solid/-vue/-svelte/-angular)webSocket(url, options)— full-duplexsubscribe+sendconnection adapter foruseChat.send()writes aRunAgentInputframe;subscribe()yields inbound chunks, ignores heartbeats, unwraps durable envelopes, and auto-reconnects a dropped durable run by reopening with?runId=&offset=(browsers can't set aLast-Event-IDhandshake header, so the offset rides in the URL).StreamReconnectLimitError) is shared with the HTTP adapters via the newcreateReconnectTracker. A fatal drop surfaces to the consumer instead of hanging.Testing
@tanstack/aistream-to-websocket 18/18;@tanstack/ai-clientconnection-adapter suites 63/63 (incl. reconnect, fatal-drop-surfacing, and open-promise-race regressions).upgradehook +ws) — ordered stream + reconnect-resume, 2/2; SSE/NDJSON regression 6/6./websocket-chatroute inexamples/ts-react-chat(type-checks + builds).docs/resumable-streams/overview + advanced + new WebSockets page; kiira-typechecked snippets, both server and client halves.Notes / follow-ups (v2, documented)
stop()does not emit an{ type: 'abort' }frame today — per-turn abort is via socket close (aborts all turns). The protocol primitive exists server-side.webSocket()connection assumes a single in-flight run at a time.🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Bug Fixes
Documentation
Tests